package com.boot.cluster.worker;

import com.alibaba.fastjson.JSON;
import com.boot.cluster.ClusterRaftClient;
import com.boot.cluster.worker.bean.*;
import com.boot.cluster.worker.runner.AbstractClusterComputeTask;
import com.boot.cluster.worker.selector.Selector;
import org.jgroups.Address;
import org.jgroups.blocks.MethodCall;
import org.jgroups.blocks.RequestOptions;
import org.jgroups.blocks.ResponseMode;
import org.jgroups.blocks.RpcDispatcher;
import org.jgroups.util.RspList;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.backoff.FixedBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;


/**
 * @author lichunwen
 */
public class ClusterScheduler extends LocalScheduler {
    private static Logger logger = LoggerFactory.getLogger(ClusterScheduler.class);
    private ClusterRaftClient client;
    private RpcDispatcher rpcDispatcher;
    private Selector<Node, TaskSplitterInfo> als;
    //用于检测非master状态时，停止本机的work状态检测任务
    //private ScheduledThreadPoolExecutor  spool;

    /***
     *
     * @param client raft客户端
     * @param als 任务分配策略
     */
    public ClusterScheduler(ClusterRaftClient client, Selector<Node, TaskSplitterInfo> als) {
        this.client = client;
        //this.startMasterCheckSchedule();
        this.rpcDispatcher = client.getRpcDispatcher();
        this.rpcDispatcher.setServerObject(new Actor(this.executor));
        this.als = als;
        //设置重试次数与重试时间间隔
        SimpleRetryPolicy policy = new SimpleRetryPolicy(3, Collections.<Class<? extends Throwable>, Boolean>singletonMap(Exception.class, true));
        // 设置重试回退操作策略，主要设置重试间隔时间
        FixedBackOffPolicy fixedBackOffPolicy = new FixedBackOffPolicy();
        fixedBackOffPolicy.setBackOffPeriod(100);
        this.retryTemplate.setRetryPolicy(policy);
        this.retryTemplate.setBackOffPolicy(fixedBackOffPolicy);
    }


    public void execute(Class<AbstractClusterComputeTask> cls) {

    }

    /***
     * 相同任务提交到所有节点,同步执行并获取执行结果
     * @param wrap
     * @return
     */
    public List<TaskResponse> submitAll(TaskWrap wrap) {
        List<TaskResponse> res = this.submit(null, wrap.getRunner(), wrap.getTasks());
        if ((res != null) && !res.isEmpty()) {
            this.combiner(res, wrap.getRunner().getCbls());
        }
        return res;

    }

    /**
     * 向目标节点批量提交任务
     *
     * @param dest
     * @param runner
     * @param infos
     * @return
     */
    private List<TaskResponse> submit(Address dest, TaskRunner runner, List<TaskSplitterInfo> infos) {
        try {
            return this.retryTemplate.execute((RetryCallback<List<TaskResponse>, Exception>) rc -> {
                MethodCall call = new MethodCall("submit", new Object[]{infos, runner}, new Class[]{List.class, TaskRunner.class});
                List<TaskResponse> rsp = this.rpcDispatcher.callRemoteMethod(dest,
                        call, new RequestOptions(ResponseMode.GET_ALL, runner.getTimeout() == null ? this.timeout : runner.getTimeout()));
                return rsp;
            });
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /***
     * 向集群批量提交任务
     * @param wrap
     * @return
     */
    @Override
    public List<TaskResponse> submit(TaskWrap wrap) {
        if ((wrap == null) || (wrap.getTasks() == null) || wrap.getTasks().isEmpty()) {
            return null;
        }
        if ((this.client.getMember() == null) || this.client.getMember().isEmpty()) {
            return null;
        }
        if (this.als == null) {
            logger.error("balance select is null");
            return null;
        }
        Map<Address, List<TaskSplitterInfo>> selects = null;// this.als.selector(this.client.getMember(), wrap.getTasks());
        if (selects.isEmpty()) {
            return null;
        }
        List<TaskResponse> res = new ArrayList<>();
        selects.forEach((k, v) -> {
            try {
                List<TaskResponse> trList = this.submit(k, wrap.getRunner(), v);
                logger.info("submit task to target: {},runner:{},Tasks:{},res:{}",
                        k.toString(), wrap.getRunner(), v, JSON.toJSONString(trList));
                if ((trList != null) && !trList.isEmpty()) {
                    res.addAll(trList);
                } else {
                    logger.info("task info :{} no result", v);
                }
            } catch (RuntimeException e) {
                logger.error("", e);
            }
        });
        if (((res != null) && !res.isEmpty()) || (wrap.getRunner().getCbls() != null)) {
            this.combiner(res, wrap.getRunner().getCbls());
        }
        return res;
    }


    @Override
    public void cancelTask(TaskRunner runner, TaskSplitterInfo info) {
        try {
            this.retryTemplate.execute((RetryCallback<Void, Exception>) rc -> {
                MethodCall call = new MethodCall("cancelTask", new Object[]{info, runner}
                        , new Class[]{TaskSplitterInfo.class, TaskRunner.class});
                this.rpcDispatcher.callRemoteMethod(null,
                        call, new RequestOptions(ResponseMode.GET_ALL,
                                runner.getTimeout() == null ? this.timeout : runner.getTimeout()));
                return null;
            });
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    @Override
    public void cancelAllTask(TaskRunner runner, List<TaskSplitterInfo> infos) {
        try {
            this.retryTemplate.execute((RetryCallback<Void, Exception>) rc -> {
                MethodCall call = new MethodCall("cancelAllTask", new Object[]{infos, runner}
                        , new Class[]{List.class, TaskRunner.class});
                this.rpcDispatcher.callRemoteMethod(null,
                        call, new RequestOptions(ResponseMode.GET_ALL,
                                runner.getTimeout() == null ? this.timeout : runner.getTimeout()));
                return null;
            });
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    @Override
    public void cancelAllTask(TaskWrap wrap) {
        try {
            this.retryTemplate.execute((RetryCallback<Void, Exception>) rc -> {
                MethodCall call = new MethodCall("cancelAllTask", new Object[]{wrap.getTasks(), wrap.getRunner()}
                        , new Class[]{List.class, TaskRunner.class});
                this.rpcDispatcher.callRemoteMethod(null,
                        call, new RequestOptions(ResponseMode.GET_ALL,
                                wrap.getRunner().getTimeout() == null ? this.timeout : wrap.getRunner().getTimeout()));
                return null;
            });
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /***
     * 向集群查询执行结果
     * @param wrap
     * @return
     */
    public List<TaskResponse> getResult(TaskWrap wrap) {
        try {
            List<TaskResponse> res = this.retryTemplate.execute((RetryCallback<List<TaskResponse>, Exception>) rc -> {
                MethodCall call = new MethodCall("getResult", new Object[]{wrap}, new Class[]{TaskWrap.class});
                RspList<TaskResponse> rsp = this.rpcDispatcher.callRemoteMethod(null,
                        call, new RequestOptions(ResponseMode.GET_ALL,
                                wrap.getRunner().getTimeout() == null ? this.timeout : wrap.getRunner().getTimeout()));
                return rsp.getResults();
            });
            if (((res != null) && !res.isEmpty()) || (wrap.getRunner().getCbls() != null)) {
                this.combiner(res, wrap.getRunner().getCbls());
            }
            return res;
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
